import io
import avro.schema
from avro.io import DatumWriter
from kafka import SimpleProducer
from kafka import KafkaClient
import json
import sys
import getopt


def send_data(bootstrap_servers,topic_name,json_path,schema_path):

    SCHEMA_PATH = schema_path

    json_file = open(json_path,"r",encoding="utf-8")
    dic_data = json_file.readline()

    # To send messages synchronously
    KAFKA = KafkaClient(bootstrap_servers)
    # Path to user.avsc avro schema


    PRODUCER = SimpleProducer(KAFKA)
    SCHEMA = avro.schema.parse(open(SCHEMA_PATH).read())

    num  =  1
    while (dic_data):
        try:
            data = json.loads(dic_data)
            writer = DatumWriter(SCHEMA)
            bytes_writer = io.BytesIO()
            encoder = avro.io.BinaryEncoder(bytes_writer)
            writer.write(data, encoder)
            raw_bytes = bytes_writer.getvalue()
            PRODUCER.send_messages(topic_name, raw_bytes)
            # print(dic_data)
        except Exception as errors:
            print(str(errors))
        dic_data = json_file.readline()
        print(num)
        num = num + 1
    json_file.close()

def main(argv):
    print(argv)
    try:
        opts, args = getopt.getopt(argv, "b:t:j:s", ["bootstrap_servers=", "topic_name=", "json_path=", "schema_path="])
    except:
        print("input is wrong ! \nexample: python .\send_lot_data.py  --bootstrap_servers 172.16.80.11:6667 --topic_name test_deye_ods_call --json_path E:/python_code_using_pycharm/little_tools/test_json.json --schema_path E:/DEYE-6.2/DEYE-6.2.3/schema/ods/ods-call.avsc")
        exit(2)
    for opt, arg in opts:
        if opt in ("-b", "--bootstrap_servers"):
            bootstrap_servers = arg
        if opt in ("-t", "--topic_name"):
            topic_name = arg
        if opt in ("-j", "--json_path"):
            json_path = arg
        if opt in ("-s", "--schema_path"):
            schema_path = arg
    send_data(bootstrap_servers,topic_name,json_path,schema_path)


if __name__ == '__main__':
    main(sys.argv[1:])
